package com.whk.push;


import java.lang.ref.SoftReference;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

public class PushUtil {

    private final AtomicBoolean sendDoor = new AtomicBoolean(false);
    private final Thread futureTask = new PushThread();
    private Queue<PushResult> pushResultQueue = new ConcurrentLinkedQueue<>();
    private Queue<SoftReference<PushMessage>> queue;
    /**
     * 执行线程池
     */
    private ExecutorService executor;
    /**
     * 简单日志记录
     */
    private PushLogger log;
    /**
     * 回调发送方法
     */
    private PushCall call;
    /**
     * 缓存结果长度
     */
    private int resultSize;

    protected PushUtil(Queue<SoftReference<PushMessage>> queue, ExecutorService executor, PushLogger log, PushCall call, int resultSize) {
        if (queue == null) {
            this.queue = new ConcurrentLinkedQueue<>();
        } else {
            this.queue = queue;
        }
        if (executor == null) {
            this.executor = Executors.newSingleThreadExecutor();
        } else {
            this.executor = executor;
        }
        if (log == null) {
            this.log = new PushLogger.DefaultPushLogger();
        } else {
            this.log = log;
        }
        if (resultSize <= 0) {
            this.resultSize = 100;
        } else {
            this.resultSize = resultSize;
        }
        if (call != null) {
            this.call = call;
        }
    }

    protected PushUtil() {
        this.queue = new ConcurrentLinkedQueue<>();
        this.executor = Executors.newSingleThreadExecutor();
        this.log = new PushLogger.DefaultPushLogger();
    }

    public Queue<PushResult> getPushResultQueue() {
        return pushResultQueue;
    }

    /**
     * 添加发送消息到队列
     *
     * @param pushMessage
     */
    public boolean putMessage(PushMessage pushMessage) {
        boolean flag = queue.add(new SoftReference<PushMessage>(pushMessage));
        if (!flag) {
            log.error(null, "添加消息发送队列失败,queue");
            return false;
        }
        if (sendDoor.compareAndSet(false, true)) { //未开启发送
            try {
                executor.execute(futureTask);
            } catch (Throwable e) {
                log.error(e, "pushUtil线程执行失败");
            }
        }
        return true;
    }

    /**
     * 开启发送
     */
    private void startSend() {
        while (!queue.isEmpty()) {
            SoftReference softReference = queue.poll();
            PushMessage pushMessage = (PushMessage) softReference.get();
            if (pushMessage == null) {
                log.error(new PushException(), "PushUtil: obj null happened");
            }
            doSend(pushMessage);
        }
        sendDoor.set(false);//发送结束，等待数据产生后重新开启
        executor.shutdown();
    }

    /**
     * 发送消息
     */
    private void doSend(PushMessage pushMessage) {
        PushResult pushResult;
        try {
            pushResult = call.call(pushMessage);
        } catch (Exception e) {
            log.error(new PushException(e), "消息推送出现错误");
            pushResult = PushResult.fail(pushMessage.getSeq());
        }
        //超出 resultSize 清理
        if (pushResultQueue.size() > resultSize) {
            pushResultQueue.clear();
        }
        pushResultQueue.add(pushResult);
    }


    /**
     * 判断待发送队列是否为空
     *
     * @return
     */
    public boolean isQueueEmpty() {
        return queue.isEmpty();
    }

    /**
     * User: Wanghk
     * Date: 15-6-5
     * Time: 下午3:28
     */
    private class PushThread extends Thread {
        @Override
        public void run() {
            startSend();
        }

//        public Boolean call() throws Exception {
//            startSend();
//            return true;
//        }
    }
}
